package com.pw.study.flink.window

import com.pw.study.flink.source.{WordBean, WordSource}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.{AllWindowFunction, WindowFunction}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ArrayBuffer

object W3 {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //1.10版本默认时间语义为ProcessingTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val ds: DataStream[WordBean] = env.addSource(new WordSource)
    //添加水印
    val dsWatermark: DataStream[WordBean] = ds.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[WordBean](Time.seconds(3)) {
      override def extractTimestamp(bean: WordBean): Long = {
        bean.ts
      }
    })
    //开启窗口
    val dsWindow: DataStream[String] = dsWatermark.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
      .apply(new AllWindowFunction[WordBean, String, TimeWindow] {
        override def apply(window: TimeWindow, beans: Iterable[WordBean], out: Collector[String]): Unit = {
          println("size: " + beans.size)
          out.collect(beans.mkString("@"))
        }
      })

    dsWindow.print("window: ")

    env.execute("window:")
  }
}
